package cn.xuexiyuan.flinkstudy.connectors;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @Description:
 *
 *  http://bahir.apache.org/docs/flink/current/flink-streaming-redis/
 *
 * @Author 左龙龙
 * @Date 21-3-24
 * @Version 1.0
 **/
public class RedisDemo {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<String> ds1 = env.socketTextStream("localhost", 6666);

        // 2.transformation
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = ds1.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
            for (String token : s.split("\\W+")) {
                collector.collect(Tuple2.of(token, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(d -> d.f0).sum(1);

        // 3.sink
        result.print();

        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").build();
        RedisSink<Tuple2<String, Integer>> redisSink = new RedisSink<Tuple2<String, Integer>>(conf, new MyRedisMapper());
        result.addSink(redisSink);

        // 4.excute
        env.execute("RedisDemo");
    }

}

class MyRedisMapper implements RedisMapper<Tuple2<String, Integer>>{

    @Override
    public RedisCommandDescription getCommandDescription() {
        // 单词统计结果存储结构 Hash   wordd_cout: {单词1: 数量1, 单词2: 数量2}
        return new RedisCommandDescription(RedisCommand.HSET, "word_count");
    }

    @Override
    public String getKeyFromData(Tuple2<String, Integer> data) {
        return data.f0;
    }

    @Override
    public String getValueFromData(Tuple2<String, Integer> data) {
        return data.f1.toString();
    }
}
